/**
 * Copyright ©2016-2020 tianyu education Corporation, All Rights Reserved
 * http://www.whty.com.cn/
 */
package profile.hujia.profile.elasticsearch.study.service;

import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchScrollRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.Scroll;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import profile.hujia.profile.elasticsearch.study.model.SourceEntity;

import java.io.BufferedWriter;
import java.io.FileWriter;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;

/**
 * class info
 *
 * @author hujia
 * @date 2019-09-10 09:17
 */
@Service
@Slf4j
public class SourceServiceImpl implements SourceService {
    @Autowired
    ObjectMapper jackson;

    @Autowired
    private RestHighLevelClient restClient;

    @Override
    public void addSource(SourceEntity sourceEntity) throws Exception {
        IndexRequest request = getIndexRequest(sourceEntity);
        IndexResponse indexResponse = restClient.index(request, RequestOptions.DEFAULT);
    }

    @Override
    public List<SourceEntity> listAll() throws Exception {
        List<SourceEntity> resultList = new ArrayList<>();
        SearchRequest searchRequest = new SearchRequest();
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.query(QueryBuilders.matchAllQuery());
        searchRequest.source(searchSourceBuilder);
        SearchResponse searchResponse = restClient.search(searchRequest, RequestOptions.DEFAULT);
        SearchHits hits = searchResponse.getHits();
        SearchHit[] searchHits = hits.getHits();
        System.out.println(hits.getTotalHits());
        for (SearchHit hit : searchHits) {
            SourceEntity sourceEntity = new SourceEntity();
            Map<String, Object> sourceAsMap = hit.getSourceAsMap();
            sourceEntity.setTitle((String) sourceAsMap.get("title"));
            sourceEntity.setGrade((String) sourceAsMap.get("grade"));
            sourceEntity.setSubject((String) sourceAsMap.get("subject"));
            sourceEntity.setArea((String) sourceAsMap.get("area"));
            sourceEntity.setType((String) sourceAsMap.get("type"));
            sourceEntity.setKind((String) sourceAsMap.get("kind"));
            sourceEntity.setYears((String) sourceAsMap.get("years"));
            sourceEntity.setPeriod((String) sourceAsMap.get("period"));

            long time = Long.valueOf(String.valueOf(sourceAsMap.get("createTime")));
            sourceEntity.setCreateTime(new Date(time));
            resultList.add(sourceEntity);
        }
        return resultList;
    }

    @Override
    public void addSourceList(List<SourceEntity> listSourceEntity) throws Exception {
        BulkRequest bulkRequest = new BulkRequest();
        int i = 0;
        for (; i < listSourceEntity.size(); ) {
            IndexRequest indexRequest = getIndexRequest(listSourceEntity.get(i));
            bulkRequest.add(indexRequest);
            ++i;
            if (i % 10000 == 0) {
                log.info("第" + i / 10000 + "次插入数据");
                restClient.bulk(bulkRequest, RequestOptions.DEFAULT);
                log.info("insert end");
                bulkRequest = new BulkRequest();
            }
        }
        if (i % 10000 != 0) {
            restClient.bulk(bulkRequest, RequestOptions.DEFAULT);
        }
    }

    private IndexRequest getIndexRequest(SourceEntity sourceEntity) throws Exception {
        String id=sourceEntity.getId();
        IndexRequest indexRequest = new IndexRequest("recommendation", "resource",id);
        String jsonString = jackson.writeValueAsString(sourceEntity);
        indexRequest.source(jsonString, XContentType.JSON);
        return indexRequest;
    }

    @Override
    public void deleteDocument(SourceEntity sourceEntity) throws Exception {
        String id=sourceEntity.getId();
        DeleteRequest request=new DeleteRequest("recommendation", "resource",id);
         restClient.delete(request, RequestOptions.DEFAULT);
    }

    @Override
    public void deleteDocument(String id) throws Exception {
        DeleteRequest request=new DeleteRequest("recommendation", "resource",id);
        restClient.delete(request, RequestOptions.DEFAULT);
    }

    @Override
    public void updateDocument(SourceEntity sourceEntity) throws Exception {
        String id=sourceEntity.getId();
        UpdateRequest request = new UpdateRequest("recommendation", "resource",id);
        String jsonString=jackson.writeValueAsString(sourceEntity);
        request.upsert(jsonString, XContentType.JSON);
        restClient.update(request, RequestOptions.DEFAULT);
    }

    @Override
    public void updateDocumentList(List<SourceEntity> listSourceEntity) throws Exception {
        BulkRequest bulkRequest = new BulkRequest();
        int i = 0;
        for (SourceEntity sourceEntity:listSourceEntity) {
            String id=sourceEntity.getId();
            UpdateRequest request = new UpdateRequest("recommendation", "resource",id);
            String jsonString=jackson.writeValueAsString(sourceEntity);
            request.upsert(jsonString, XContentType.JSON);
            bulkRequest.add(request);
            ++i;
            if (i % 10000 == 0) {
                log.info("第" + i / 10000 + "次插入数据");
                restClient.bulk(bulkRequest, RequestOptions.DEFAULT);
                log.info("insert end");
                bulkRequest = new BulkRequest();
            }
        }
        if (i % 10000 != 0) {
            restClient.bulk(bulkRequest, RequestOptions.DEFAULT);
        }
    }

    @Override
    public void saveToTxt(String filePath) throws Exception {
        final Scroll scroll = new Scroll(TimeValue.timeValueMinutes(1L));
        SearchRequest searchRequest = new SearchRequest();
        searchRequest.scroll(scroll);
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.query(QueryBuilders.matchAllQuery());
        searchSourceBuilder.size(10000);
        searchRequest.source(searchSourceBuilder);
        SearchResponse searchResponse = restClient.search(searchRequest, RequestOptions.DEFAULT);
        String scrollId = searchResponse.getScrollId();
        SearchHit[] searchHits = searchResponse.getHits().getHits();
        BufferedWriter out = new BufferedWriter(new FileWriter(filePath, true));
        long count = 0;
        long cn = 1;
        //把导出的结果以JSON的格式写到文件里
        while (searchHits != null && searchHits.length > 0) {
            for (SearchHit hit : searchHits) {
                String json = hit.getSourceAsString();
                if (!json.isEmpty() && !"".equals(json)) {
                    out.write(json);
                    out.write("\r\n");
                    count++;
                }
            }
            SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId);
            scrollRequest.scroll(scroll);
            searchResponse = restClient.scroll(scrollRequest, RequestOptions.DEFAULT);
            scrollId = searchResponse.getScrollId();
            searchHits = searchResponse.getHits().getHits();
            System.out.println("处理完成第" + cn + "批数据");
            cn++;
        }
        System.out.println("总共写入数据:" + count);
        out.close();
    }

    @Override
    public SourceEntity getDocument(String id) throws Exception {
        GetRequest request = new GetRequest("recommendation", "resource", id);
        GetResponse response = restClient.get(request);
        String source = response.getSourceAsString();
        SourceEntity sourceEntity = jackson.readValue(source, SourceEntity.class);
        return sourceEntity;
    }
}
